package concurrent.completableFuture;

import org.junit.Test;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;

/**
 * 异步回调
 *
 * @author LoneRanger
 * @version 1.0
 */
public class AsyncCallBackTest {

	/**
	 * thenApply 表示某个任务执行完成后执行的动作，即回调方法，会将该任务的执行结果即方法返回值作为入参传递到回调方法中
	 * thenApplyAsync与thenApply的区别在于，前者是将job2提交到线程池中异步执行，实际执行job2的线程可能是另外一个线程，
	 * 后者是由执行job1的线程立即执行job2，即两个job都是同一个线程执行的。
	 *
	 * @throws ExecutionException
	 * @throws InterruptedException
	 */
	@Test
	public void test0() throws ExecutionException, InterruptedException {
		ForkJoinPool pool = new ForkJoinPool();
		// 创建异步执行任务:
		CompletableFuture<Double> cf = CompletableFuture.supplyAsync(() -> {
			/*
			 * job1
			 */
			System.out.println(Thread.currentThread() + " start job1,time->" + System.currentTimeMillis());
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
			}
			System.out.println(Thread.currentThread() + " exit job1,time->" + System.currentTimeMillis());
			return 1.2;
		}, pool);

		//cf关联的异步任务的返回值作为方法入参，传入到thenApply的方法中
		//thenApply这里实际创建了一个新的CompletableFuture实例
		CompletableFuture<String> cf2 = cf.thenApply((result) -> {
			/*
			 * job2
			 */
			System.out.println(Thread.currentThread() + " start job2,time->" + System.currentTimeMillis());
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
			}
			System.out.println(Thread.currentThread() + " exit job2,time->" + System.currentTimeMillis());
			return "test:" + result;
		});
		System.out.println("main concurrent.thread start cf.get(),time->" + System.currentTimeMillis());
		//等待子任务执行完成
		System.out.println("run result->" + cf.get());
		System.out.println("main concurrent.thread start cf2.get(),time->" + System.currentTimeMillis());
		System.out.println("run result->" + cf2.get());
		System.out.println("main concurrent.thread exit,time->" + System.currentTimeMillis());
	}

	/**
	 * thenAccept / thenRun
	 * thenAccept 同 thenApply 接收上一个任务的返回值作为参数，但是无返回值；
	 * thenRun 的方法没有入参，也买有返回值
	 */
	@Test
	public void test1() throws ExecutionException, InterruptedException {
		ForkJoinPool pool = new ForkJoinPool();
		// 创建异步执行任务:
		CompletableFuture<Double> cf = CompletableFuture.supplyAsync(() -> {
			System.out.println(Thread.currentThread() + " start job1,time->" + System.currentTimeMillis());
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
			}
			System.out.println(Thread.currentThread() + " exit job1,time->" + System.currentTimeMillis());
			return 1.2;
		}, pool);
		//cf关联的异步任务的返回值作为方法入参，传入到thenApply的方法中
		CompletableFuture cf2 = cf.thenApply((result) -> {
			System.out.println(Thread.currentThread() + " start job2,time->" + System.currentTimeMillis());
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
			}
			System.out.println(Thread.currentThread() + " exit job2,time->" + System.currentTimeMillis());
			return "test:" + result;
		}).thenAccept((result) -> {
			//接收上一个任务的执行结果作为入参，但是没有返回值
			System.out.println(Thread.currentThread() + " start job3,time->" + System.currentTimeMillis());
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
			}
			System.out.println(result);
			System.out.println(Thread.currentThread() + " exit job3,time->" + System.currentTimeMillis());
		}).thenRun(() -> {
			//无入参，也没有返回值
			System.out.println(Thread.currentThread() + " start job4,time->" + System.currentTimeMillis());
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
			}
			System.out.println("thenRun do something");
			System.out.println(Thread.currentThread() + " exit job4,time->" + System.currentTimeMillis());
		});
		System.out.println("main concurrent.thread start cf.get(),time->" + System.currentTimeMillis());
		//等待子任务执行完成
		System.out.println("run result->" + cf.get());
		System.out.println("main concurrent.thread start cf2.get(),time->" + System.currentTimeMillis());
		//cf2 等待最后一个thenRun执行完成
		System.out.println("run result->" + cf2.get());
		System.out.println("main concurrent.thread exit,time->" + System.currentTimeMillis());
	}

	/**
	 * exceptionally
	 * exceptionally方法指定某个任务执行异常时执行的回调方法，会将抛出异常作为参数传递到回调方法中，
	 * 如果该任务正常执行则会exceptionally方法返回的CompletionStage的result就是该任务正常执行的结果
	 */
	@Test
	public void test3() throws ExecutionException, InterruptedException {
		ForkJoinPool pool = new ForkJoinPool();
		// 创建异步执行任务:
		CompletableFuture<Double> cf = CompletableFuture.supplyAsync(() -> {
			System.out.println(Thread.currentThread() + "job1 start,time->" + System.currentTimeMillis());
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
			}
			if (true) {
				throw new RuntimeException("test");
			} else {
				System.out.println(Thread.currentThread() + "job1 exit,time->" + System.currentTimeMillis());
				return 1.2;
			}
		}, pool);
		//cf执行异常时，将抛出的异常作为入参传递给回调方法
		CompletableFuture<Double> cf2 = cf.exceptionally((param) -> {
			System.out.println(Thread.currentThread() + " start,time->" + System.currentTimeMillis());
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
			}
			System.out.println("error stack trace->");
			param.printStackTrace();
			System.out.println(Thread.currentThread() + " exit,time->" + System.currentTimeMillis());
			return -1.1;
		});
		//cf正常执行时执行的逻辑，如果执行异常则不调用此逻辑
		CompletableFuture cf3 = cf.thenAccept((param) -> {
			System.out.println(Thread.currentThread() + "job2 start,time->" + System.currentTimeMillis());
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
			}
			System.out.println("param->" + param);
			System.out.println(Thread.currentThread() + "job2 exit,time->" + System.currentTimeMillis());
		});
		System.out.println("main concurrent.thread start,time->" + System.currentTimeMillis());
		// 等待子任务执行完成,此处无论是job2和job3都可以实现job2退出，主线程才退出，
		// 如果是cf，则主线程不会等待job2执行完成自动退出了
		// cf2.get时，没有异常，但是依然有返回值，就是cf的返回值
		System.out.println("run result->" + cf2.get());
		System.out.println("main concurrent.thread exit,time->" + System.currentTimeMillis());
	}

	/**
	 * whenComplete
	 * whenComplete是当某个任务执行完成后执行的回调方法，会将执行结果或者执行期间抛出的异常传递给回调方法，如果是正常执行则异常为null
	 * ，回调方法对应的CompletableFuture的result和该任务一致，如果该任务正常执行，则get方法返回执行结果，如果是执行异常，则get方法抛出异常
	 */
	@Test
	public void test4() throws ExecutionException, InterruptedException {
		// 创建异步执行任务:
		CompletableFuture<Double> cf = CompletableFuture.supplyAsync(() -> {
			System.out.println(Thread.currentThread() + "job1 start,time->" + System.currentTimeMillis());
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
			}
			if (false) {
				throw new RuntimeException("test");
			} else {
				System.out.println(Thread.currentThread() + "job1 exit,time->" + System.currentTimeMillis());
				return 1.2;
			}
		});
		//cf执行完成后会将执行结果和执行过程中抛出的异常传入回调方法，如果是正常执行的则传入的异常为null
		CompletableFuture<Double> cf2 = cf.whenComplete((result, throwable) -> {
			System.out.println(Thread.currentThread() + "job2 start,time->" + System.currentTimeMillis());
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
			}
			if (throwable != null) {
				System.out.println("error stack trace->");
				throwable.printStackTrace();
			} else {
				System.out.println("run succ,result->" + result);
			}
			System.out.println(Thread.currentThread() + "job2 exit,time->" + System.currentTimeMillis());
		});
		//等待子任务执行完成
		System.out.println("main concurrent.thread start wait,time->" + System.currentTimeMillis());
		//如果cf是正常执行的，cf2.get的结果就是cf执行的结果
		//如果cf是执行异常，则cf2.get会抛出异常
		System.out.println("run result->" + cf2.get());
		System.out.println("main concurrent.thread exit,time->" + System.currentTimeMillis());

	}

}
